package com.hucais.sync.es2hive.service

import com.hucais.core.constant.Constants
import com.hucais.core.utils.CommonUtils
import com.hucais.etl.common.service.CommonQueryService
import com.hucais.etl.history.bean.{HistoryKfzBookStore, TmpKfzBookStore}
import com.hucais.etl.history.dao.HiveDao
import org.apache.spark.SparkContext
import org.apache.spark.sql.{Dataset, SaveMode, SparkSession}

object SyncKfzService {
  def action(ssc: SparkContext, sparkSession: SparkSession): Unit = {
    val hdfsPath: String = "/hucaisdata/published/bak/book_store_kfz/"
    val storeDS: Dataset[HistoryKfzBookStore] = HiveDao.getKfzBookStore(sparkSession)
    storeDS.cache()
    val tmpStroeDS = cleanStoreESData(sparkSession, storeDS)
    CommonQueryService.saveAsFileAbsPath(tmpStroeDS.toDF(), hdfsPath, Constants.HIVE_SPLIT_STR, SaveMode.Overwrite)
  }

  /**
   * 清洗ES中store的数据
   *
   * @param sparkSession
   * @param storeDS store的index数据
   * @return
   */
  private def cleanStoreESData(sparkSession: SparkSession, storeDS: Dataset[HistoryKfzBookStore]): Dataset[TmpKfzBookStore] = {
    import sparkSession.implicits._
    storeDS.mapPartitions(partitions => {
      partitions
        .filter(item => CommonUtils.isNotBlankExt(item.isbn) && CommonUtils.isNotBlankExt(item.name_of_shop) &&
          CommonUtils.isNotBlankExt(item.store_pricing) && CommonUtils.isNotBlankExt(item.selling_price))
        .map(item => {
          val isbn = CommonUtils.handleBlankStr(item.isbn)
          val storeName = CommonUtils.handleBlankStr(item.name_of_shop)
          val bookNum = CommonUtils.handleBlankNumber(item.number)
          val storePricing = CommonUtils.handleBlankFloat(CommonUtils.filterSomeStr(item.store_pricing))
          val sellingPrince = CommonUtils.handleBlankFloat(item.selling_price)
          val slogan = CommonUtils.filterSomeStr(CommonUtils.handleBlankStr(item.slogan))
          val bookComments = CommonUtils.handleBlankNumber(item.book_review)
          val conditions = CommonUtils.handleBlankStr(item.conditions)

          TmpKfzBookStore(isbn, storeName, slogan, bookComments, sellingPrince, storePricing, conditions, bookNum)
        })
    })
  }
}
